#include "sms_service.h"
#include <sys/mman.h>
#include <fcntl.h>
#include <unistd.h>
#include <regex>
#include <random>


#define BUFF_SIZE 1024 * 128


namespace sms {


Service::Service
(
    std::string request_url,
    std::string request_type,
    std::string response_type,
    std::function<nlohmann::json(nlohmann::json)> callable,
    std::string ip,
    int port
)
{
    this->_request_url = request_url;
    this->_request_type = request_type;
    this->_response_type = response_type;
    this->_callable = callable;

    this->_ip = ip;
    this->_port = port;

    nlohmann::json all_types = load_msg_types();
    if (!all_types.contains(request_type))
    {
        std::cerr << "[SMS_ERROR]: (" << request_type << ") " << ec2str(227) << std::endl;
        std::abort();
    }
    if (!all_types.contains(response_type))
    {
        std::cerr << "[SMS_ERROR]: (" << response_type << ") " << ec2str(227) << std::endl;
        std::abort();
    }
    int url_err = check_topic_url(request_url);
    if (url_err != 0)
    {
        std::cerr << "[SMS_ERROR]: (" << request_url << ") " << ec2str(url_err) << std::endl;
        std::abort();
    }

    this->_last_send_time = 0.0;
    this->_force_quit = false;
    this->_heartbeat_running = false;
    this->_running = true;

    this->_client_socket = -1;
    this->_buf = new char[BUFF_SIZE + 1];
    this->_recv_t = nullptr;
    this->_send_t = nullptr;
    this->_use_shm = false;

    if (_startswith(request_type, "memory_msgs::"))
    {
        std::regex pattern("/");
        this->_shm_name = "/" + std::regex_replace(request_type, pattern, "_");
        this->_use_shm = true;
    }

    this->_recv_t = new std::thread(&Service::recv_loop, this);
    this->_send_t = new std::thread(&Service::send_loop, this);

    this->_link();
}

Service::~Service()
{
    this->_force_quit = true;
    this->join();
    if (this->_buf)
        delete this->_buf;
    if (this->_recv_t)
        delete this->_recv_t;
    if (this->_send_t)
        delete this->_send_t;
}

void Service::join()
{
    if (this->_recv_t)
        this->_recv_t->join();
    if (this->_send_t)
        this->_send_t->join();
}

void Service::kill()
{
    if (!this->_force_quit)
    {
        this->_force_quit = true;
        this->_running = false;
        this->_close_socket();
    }
}

void Service::recv_loop()
{
    while (this->_running)
    {
        if (this->_force_quit)
        {
            break;
        }
        if (this->_client_socket < 0)
        {
            this->_heartbeat_running = false;
            sleep(1);
            continue;
        }
        ssize_t buf_len = read(this->_client_socket, this->_buf, BUFF_SIZE);
        if (buf_len <= 0)
        {
            this->_close_socket();
            this->_heartbeat_running = false;
            sleep(1);
            continue;
        }
        this->_buf[buf_len] = 0;
        std::string data(this->_buf, buf_len);

        std::vector<std::string> checked_msgs;
        std::vector<std::string> parted_msgs;
        std::vector<int> parted_lens;
        std::vector<std::string> recv_msgs;

        _check_msg(data, checked_msgs, parted_msgs, parted_lens);
        if (parted_msgs.size() > 0)
        {
            for (int i=0; i<parted_msgs.size(); i++)
            {
                if (parted_lens[i] > 0)
                {
                    this->_last_msg = parted_msgs[i];
                    this->_last_msg_len = parted_lens[i];
                }
                else
                {
                    this->_last_msg.append(parted_msgs[i]);
                    if (this->_last_msg_len > 0 && this->_last_msg_len <= this->_last_msg.size())
                    {
                        recv_msgs.push_back(this->_last_msg.substr(0, this->_last_msg_len));
                        this->_last_msg_len = 0;
                        this->_last_msg.clear();
                    }
                }
            }
        }
        for (int i=0; i<checked_msgs.size(); i++)
            recv_msgs.push_back(checked_msgs[i]);

        if (recv_msgs.size() > 0)
        {
            for (int i=0; i<recv_msgs.size(); i++)
                this->_parse_msg(recv_msgs[i]);
        }
    }
}

void Service::send_loop()
{
    std::random_device rd;
    std::mt19937 rng(rd());
    std::uniform_int_distribution<int> dist(1, 3);

    int n_try = 0;
    while (this->_running)
    {
        while (this->_heartbeat_running)
        {
            if (this->_force_quit)
            {
                break;
            }
            if (this->_client_socket < 0)
            {
                this->_heartbeat_running = false;
            }
            else
            {
                if (get_time_sec() - this->_last_send_time >= 2.0)
                {
                    this->_heartbeat();
                }
            }
            sleep(dist(rng));
        }
        if (this->_force_quit)
        {
            break;
        }
        sleep(1);
        if (!this->_heartbeat_running)
        {
            n_try++;
            if (n_try > 5)
            {
                n_try = 0;
                this->_link();
            }
        }
    }
}


void Service::_heartbeat()
{
    nlohmann::json heartbeat_msg = def_msg("_sys_msgs::Service");
    heartbeat_msg["request_type"] = this->_request_type;
    heartbeat_msg["response_type"] = this->_response_type;
    heartbeat_msg["url"] = this->_request_url;
    std::string bytes = encode_msg(heartbeat_msg);

    this->_send_mtx.lock();
    ssize_t ret = send(this->_client_socket, bytes.c_str(), bytes.size(), MSG_NOSIGNAL | MSG_DONTWAIT);
    this->_send_mtx.unlock();

    this->_last_send_time = get_time_sec();
}

void Service::_close_socket()
{
	if (this->_client_socket > 0)
    {
		close(this->_client_socket);
	}
	this->_client_socket = -1;
}

bool Service::_link()
{
    if (this->_client_socket != -1)
    {
		this->_close_socket();
	}
	this->_client_socket = socket(PF_INET , SOCK_STREAM , 0);
	if (this->_client_socket == -1)
    {
		return false;
	}

    memset(&this->_server_addr , 0 , sizeof(this->_server_addr));
	this->_server_addr.sin_family = AF_INET;
	this->_server_addr.sin_port = htons(this->_port);
    inet_pton(AF_INET, this->_ip.c_str(), &this->_server_addr.sin_addr);

    struct timeval timeout;
    timeout.tv_sec = 5;
    timeout.tv_usec = 0;
    if (setsockopt(this->_client_socket, SOL_SOCKET, SO_RCVTIMEO, &timeout, sizeof(timeout)) < 0)
    {
        this->_close_socket();
		return false;
    }

    if (connect(this->_client_socket, (struct sockaddr*)&this->_server_addr, sizeof(this->_server_addr)) == -1)
    {
		this->_close_socket();
		return false;
	}

    this->_last_msg_len = 0;
    this->_last_msg.clear();
    this->_heartbeat();
    this->_heartbeat_running = true;
    return true;
}


void Service::_parse_msg(std::string msg)
{
    nlohmann::json json_msg;
    if (decode_msg(msg, json_msg))
    {
        if (json_msg["type"] == "_sys_msgs::Request")
        {
            nlohmann::json resp_msg = this->_callable(json_msg["msg"]);
            if (resp_msg.contains("timestamp") && resp_msg["timestamp"] == 0.0)
            {
                resp_msg["timestamp"] = get_time_sec();
            }
            if (resp_msg.contains("type") && resp_msg["type"] == this->_response_type)
            {
                nlohmann::json response = sms::def_msg("_sys_msgs::Response");
                response["id"] = json_msg["id"];
                response["client_key"] = json_msg["client_key"];
                response["msg"] = resp_msg;

                std::string bytes = encode_msg(response);

                this->_send_mtx.lock();
                ssize_t ret = send(this->_client_socket, bytes.c_str(), bytes.size(), MSG_NOSIGNAL | MSG_DONTWAIT);
                this->_send_mtx.unlock();

                this->_last_send_time = get_time_sec();
            }
            else if (resp_msg.contains("type"))
            {
                std::cout << "[SMS_ERROR]: (" << resp_msg["type"].get<std::string>() << 
                    " != " << this->_response_type << ") " << ec2str(240) << std::endl;
            }
        }
        else if (json_msg["type"] == "_sys_msgs::Result")
        {
            if (json_msg["error_code"] > 0)
            {
                std::cout << "[SMS_ERROR]: " << ec2str(json_msg["error_code"]) << std::endl;
            }
        }
    }
}


nlohmann::json Service::cvimg2sms_mem(const cv::Mat& cvimg, const std::string frame_id)
{
    if (this->_use_shm)
    {
        return sms::cvimg2sms_mem(cvimg, this->_shm_name, frame_id);
    }
    else
    {
        std::cout << "[SMS_ERROR]: " << ec2str(501) << std::endl;
        return {};
    }
}


}
